package net.opentsdb.plugin.listener;

import com.google.common.base.Joiner;
import com.google.common.collect.Sets;
import net.opentsdb.core.TSDB;
import net.opentsdb.core.TsuidExchange;
import net.opentsdb.tsd.controller.RemoveTagsController;
import net.opentsdb.uid.UniqueId;
import net.opentsdb.utils.expiringmap.ExpirationListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class TsuidExpiringListener
        implements ExpirationListener {

    private static final Logger LOG = LoggerFactory.getLogger(TsuidExpiringListener.class);

    private TSDB tsdb;
    private Set<String> set;
    private static volatile ScheduledExecutorService expirer;

    public TsuidExpiringListener(TSDB tsdb){
        super();
        this.tsdb = tsdb;
        set = Sets.newConcurrentHashSet();
        expirer = Executors.newSingleThreadScheduledExecutor();
    }

    @Override
    public void expired(Object key, Object value) {

        byte[] bytes = ((ByteBuffer)key).array();
        // Map里的数据已经删掉，应该和手动删除区分开
        tsdb.deleteMapTsuid(UniqueId.uidToString(bytes), false);

        if(set.size() == 0){
            LOG.debug("set size is zero, so schedule clear tsuids and tags");
            scheduleClearSet();
        }

        String tsuid = UniqueId.uidToString(bytes);
//        LOG.debug("tsuid {} ttl end, set size {},set contains {}, so delete it",tsuid,set.size(),set.contains(tsuid));
        set.add(tsuid);
    }

    private void scheduleClearSet(){
        expirer.schedule(() -> {
            List<String> tsuids;
            Set<String> metrics = Sets.newHashSet();

            LOG.debug("待清除set的大小:{}",set.size());

            synchronized (TsuidExpiringListener.class){
                tsuids = new ArrayList<>(set);
                set.clear();
            }
            // 批量删除时间线
            try{
                tsdb.getHbaseWrapper().metaUtils.deleteTsuids(tsuids);
            }catch (Exception e){
                LOG.error("ttl delete tsuids {} error",Joiner.on(",").join(tsuids),e);
            }

            tsuids.forEach(tsuid -> metrics.add(TsuidExchange.getMetricFromTSUID(tsdb, UniqueId.stringToUid(tsuid))));

            // 清理tags信息
            metrics.forEach(metric -> {
                try{
                    (new RemoveTagsController(tsdb, metric)).handle();
                }catch (Exception e){
                    LOG.error("ttl Remove tags error",Joiner.on(",").join(metrics),e);
                }
            });

            // 清理完时间线还有新数据
            if(set.size() > 0) {
                LOG.debug("set size {}, so delete it again",set.size());
                scheduleClearSet();
            }
        }, 5,TimeUnit.MINUTES);
    }
}
